package org.hhf.pubsub;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.regex.Pattern;

/**
 * PubSubRedis类实现了Redis的发布订阅功能
 */
public class PubSubRedis {
    // 存储频道订阅者
    private final ConcurrentHashMap<String, Set<Subscriber>> channelSubscribers = new ConcurrentHashMap<>();
    // 存储模式订阅者
    private final ConcurrentHashMap<String, Set<Subscriber>> patternSubscribers = new ConcurrentHashMap<>();
    
    /**
     * 订阅者接口
     */
    public interface Subscriber {
        void onMessage(String channel, String message);
        void onPatternMessage(String pattern, String channel, String message);
        String getId();
    }
    
    /**
     * 订阅频道
     */
    public synchronized long subscribe(Subscriber subscriber, String... channels) {
        long count = 0;
        for (String channel : channels) {
            Set<Subscriber> subscribers = channelSubscribers.computeIfAbsent(channel, 
                k -> new CopyOnWriteArraySet<>());
            if (subscribers.add(subscriber)) {
                count++;
            }
        }
        return count;
    }
    
    /**
     * 取消订阅频道
     */
    public synchronized long unsubscribe(Subscriber subscriber, String... channels) {
        long count = 0;
        if (channels.length == 0) {
            // 取消所有频道订阅
            for (Set<Subscriber> subscribers : channelSubscribers.values()) {
                if (subscribers.remove(subscriber)) {
                    count++;
                }
            }
        } else {
            for (String channel : channels) {
                Set<Subscriber> subscribers = channelSubscribers.get(channel);
                if (subscribers != null && subscribers.remove(subscriber)) {
                    count++;
                    if (subscribers.isEmpty()) {
                        channelSubscribers.remove(channel);
                    }
                }
            }
        }
        return count;
    }
    
    /**
     * 订阅模式
     */
    public synchronized long psubscribe(Subscriber subscriber, String... patterns) {
        long count = 0;
        for (String pattern : patterns) {
            Set<Subscriber> subscribers = patternSubscribers.computeIfAbsent(pattern, 
                k -> new CopyOnWriteArraySet<>());
            if (subscribers.add(subscriber)) {
                count++;
            }
        }
        return count;
    }
    
    /**
     * 取消订阅模式
     */
    public synchronized long punsubscribe(Subscriber subscriber, String... patterns) {
        long count = 0;
        if (patterns.length == 0) {
            // 取消所有模式订阅
            for (Set<Subscriber> subscribers : patternSubscribers.values()) {
                if (subscribers.remove(subscriber)) {
                    count++;
                }
            }
        } else {
            for (String pattern : patterns) {
                Set<Subscriber> subscribers = patternSubscribers.get(pattern);
                if (subscribers != null && subscribers.remove(subscriber)) {
                    count++;
                    if (subscribers.isEmpty()) {
                        patternSubscribers.remove(pattern);
                    }
                }
            }
        }
        return count;
    }
    
    /**
     * 发布消息到频道
     */
    public long publish(String channel, String message) {
        long count = 0;
        
        // 发送给频道订阅者
        Set<Subscriber> channelSubs = channelSubscribers.get(channel);
        if (channelSubs != null) {
            for (Subscriber subscriber : channelSubs) {
                try {
                    subscriber.onMessage(channel, message);
                    count++;
                } catch (Exception e) {
                    // 忽略订阅者处理异常
                }
            }
        }
        
        // 发送给模式订阅者
        for (Map.Entry<String, Set<Subscriber>> entry : patternSubscribers.entrySet()) {
            String pattern = entry.getKey();
            if (matchPattern(pattern, channel)) {
                for (Subscriber subscriber : entry.getValue()) {
                    try {
                        subscriber.onPatternMessage(pattern, channel, message);
                        count++;
                    } catch (Exception e) {
                        // 忽略订阅者处理异常
                    }
                }
            }
        }
        
        return count;
    }
    
    /**
     * 检查频道是否匹配模式
     */
    private boolean matchPattern(String pattern, String channel) {
        // 简单的glob模式匹配
        String regex = pattern.replace("*", ".*").replace("?", ".");
        return Pattern.matches(regex, channel);
    }
    
    /**
     * 获取频道的订阅者数量
     */
    public long getChannelSubscriberCount(String channel) {
        Set<Subscriber> subscribers = channelSubscribers.get(channel);
        return subscribers != null ? subscribers.size() : 0;
    }
    
    /**
     * 获取所有频道
     */
    public Set<String> getChannels() {
        return new HashSet<>(channelSubscribers.keySet());
    }
    
    /**
     * 获取所有模式
     */
    public Set<String> getPatterns() {
        return new HashSet<>(patternSubscribers.keySet());
    }
    
    /**
     * 获取订阅者的订阅信息
     */
    public Map<String, Object> getSubscriberInfo(Subscriber subscriber) {
        Map<String, Object> info = new HashMap<>();
        
        // 统计频道订阅
        Set<String> subscribedChannels = new HashSet<>();
        for (Map.Entry<String, Set<Subscriber>> entry : channelSubscribers.entrySet()) {
            if (entry.getValue().contains(subscriber)) {
                subscribedChannels.add(entry.getKey());
            }
        }
        
        // 统计模式订阅
        Set<String> subscribedPatterns = new HashSet<>();
        for (Map.Entry<String, Set<Subscriber>> entry : patternSubscribers.entrySet()) {
            if (entry.getValue().contains(subscriber)) {
                subscribedPatterns.add(entry.getKey());
            }
        }
        
        info.put("channels", subscribedChannels);
        info.put("patterns", subscribedPatterns);
        info.put("total", subscribedChannels.size() + subscribedPatterns.size());
        
        return info;
    }
}